package com.lzq.mq.broker.config;

import com.alibaba.fastjson2.JSON;
import com.alibaba.fastjson2.JSONWriter;
import com.lzq.mq.broker.cache.CommonCache;
import com.lzq.mq.broker.model.ConsumeQueueOffsetModel;
import com.lzq.mq.broker.utils.FileContentUtil;
import com.lzq.mq.common.constants.BrokerConstants;
import io.netty.util.internal.StringUtil;
import lombok.extern.slf4j.Slf4j;

import java.util.concurrent.TimeUnit;

/**
 * 消费队列偏移量加载器
 */
@Slf4j
public class ConsumeQueueOffsetLoader {

    /**
     * 加载消费队列偏移量到CommonCache
     */
    public static void loadProperties() {
        GlobalProperties globalProperties = CommonCache.getGlobalProperties();
        String basePath = globalProperties.getBrokerHome();
        if (StringUtil.isNullOrEmpty(basePath)) {
            throw new IllegalArgumentException("非法的MQ_HOME");
        }
        String consumeQueueOffsetJsonFilePath = basePath + BrokerConstants.CONSUME_QUEUE_OFFSET_JSON_PATH;
        // 加载JSON, 读取存在的主题
        String s = FileContentUtil.readFromFile(consumeQueueOffsetJsonFilePath);
        ConsumeQueueOffsetModel consumeQueueOffsetModel = JSON.parseObject(s, ConsumeQueueOffsetModel.class);
        // 添加映射
        CommonCache.setConsumeQueueOffsetModel(consumeQueueOffsetModel);
    }

    /**
     * 启动主题信息刷盘定时任务
     */
    public void startRefreshConsumeQueueOffsetTask() {
        // 异步线程
        // 每15s将内存中的配置刷新的磁盘中
        // 类似与redis的rdb持久化思路
        CommonThreadPoolConfig.refreshConsumeQueueOffsetExecutor.execute(() -> {
            do {
                try {
                    // 先sleep, 防止初次加载时不必要的刷盘
                    TimeUnit.SECONDS.sleep(BrokerConstants.REFRESH_CONSUME_QUEUE_OFFSET_TIME_STEP);
                    // 刷盘
                    String consumeQueueOffsetJsonFilePath = CommonCache.getGlobalProperties().getBrokerHome() + BrokerConstants.CONSUME_QUEUE_OFFSET_JSON_PATH;
                    ConsumeQueueOffsetModel consumeQueueOffsetModel = CommonCache.getConsumeQueueOffsetModel();
                    FileContentUtil.overWriteToFile(consumeQueueOffsetJsonFilePath, JSON.toJSONString(consumeQueueOffsetModel, JSONWriter.Feature.PrettyFormat));
                    //log.info("消费队列偏移信息刷盘...");
                } catch (InterruptedException e) {
                    throw new RuntimeException(e);
                }
            } while (true);
        });
    }
}
